package com.yomob.kylin.facility;

import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;

import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.spark.sql.Column;
import org.apache.spark.sql.functions;

import com.google.common.base.Joiner;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;


public class GroupByUtils {
    private static final List<String> baseDimensions = Lists.newArrayList("date", "appid");

    public static List<List<Column>> getGroupByColumnCombinations(String... dimensions) {
        Set<String> input = Sets.newHashSet(dimensions);
        input.removeAll(baseDimensions);
        Set<Set<String>> sets = Sets.powerSet(input);
        List<List<Column>> results = Lists.newArrayList();
        sets.forEach(set -> {
            List<String> list = Lists.newArrayList(set);
            list.addAll(baseDimensions);
            list.sort(String::compareTo);
            List<Column> columns = Lists.newArrayList();
            columns.addAll(list.stream().map(functions::col).collect(Collectors.toList()));
            results.add(columns);
        });
        return results;
    }

    public static List<List<String>> getGroupByFieldCombinations(String... dimensions) {
        Set<String> input = Sets.newHashSet(dimensions);
        input.removeAll(baseDimensions);
        Set<Set<String>> sets = Sets.powerSet(input);
        List<List<String>> results = Lists.newArrayList();
        sets.forEach(set -> {
            List<String> list = Lists.newArrayList(set);
            list.addAll(baseDimensions);
            list.sort(String::compareTo);
            results.add(list);
        });
        return results;
    }

    public static String getCollectionName(String collectionNamePrefix, List<String> groupByFields) {
        if (CollectionUtils.isEmpty(groupByFields)) {
            return collectionNamePrefix;
        }
        List<String> groupBy = Lists.newArrayList(groupByFields);
        groupBy.removeAll(baseDimensions);
        String suffix = Joiner.on("#").skipNulls().join(groupBy);
        if (StringUtils.isBlank(suffix)) {
            return collectionNamePrefix;
        }
        return collectionNamePrefix + "__groupby_" + suffix;
    }
}
